package com.example.order.service.kafka;

/**
 * @author FanFu
 * @version 1.0
 * @date 2025/11/8 20:52
 */

import com.example.common.model.kafka.UserMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class UserKafkaConsumer {

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 监听用户注册事件
     */
    @KafkaListener(topics = "${app.kafka.topics.user-register}")
    public void consumeUserRegister(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String messageJson = record.value();
            UserMessage userMessage = objectMapper.readValue(messageJson, UserMessage.class);

            log.info("收到用户注册消息: {}", userMessage);

            // 这里可以处理用户注册后的业务逻辑
            // 比如：发送欢迎邮件、初始化用户积分等

            // 手动确认消息
            ack.acknowledge();
            log.info("用户注册消息处理完成: {}", userMessage.getUserId());

        } catch (Exception e) {
            log.error("处理用户注册消息失败: {}", record.value(), e);
        }
    }

    /**
     * 监听用户更新事件
     */
    @KafkaListener(topics = "${app.kafka.topics.user-update}")
    public void consumeUserUpdate(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String messageJson = record.value();
            UserMessage userMessage = objectMapper.readValue(messageJson, UserMessage.class);

            log.info("收到用户更新消息: {}", userMessage);

            // 这里可以处理用户更新后的业务逻辑
            // 比如：更新缓存、同步用户信息到其他系统等

            // 手动确认消息
            ack.acknowledge();
            log.info("用户更新消息处理完成: {}", userMessage.getUserId());

        } catch (Exception e) {
            log.error("处理用户更新消息失败: {}", record.value(), e);
        }
    }
}
